package org.jetlinks.pro.device.message;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.device.DeviceOperator;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.message.Headers;
import org.jetlinks.core.metadata.DeviceMetadata;
import org.jetlinks.core.metadata.PropertyMetadata;
import org.jetlinks.pro.PropertyConstants;
import org.jetlinks.pro.device.entity.DeviceLatestData;
import org.jetlinks.pro.device.service.data.DeviceLatestDataService;
import org.jetlinks.pro.gateway.DeviceMessageUtils;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import java.util.HashMap;
import java.util.Map;

/**
 * 合并最新的数据到消息中
 *
 * @author zhouhao
 * @since 1.12
 */
@Component
@AllArgsConstructor
@Slf4j
public class MergeLatestPropertyInterceptor implements DeviceMessagePublishInterceptor {

    private final DeviceLatestDataService dataService;

    @Override
    public boolean isSupport(DeviceMessage message) {
        return message.getHeaderOrElse(Headers.mergeLatest, () -> false);
    }

    @Override
    public Mono<DeviceMessage> beforePublish(DeviceOperator device, DeviceMessage msg) {
        String productId = msg.getHeader(PropertyConstants.productId).orElse("null");
        return Mono
            .zip(
                Mono.justOrEmpty(DeviceMessageUtils.tryGetProperties(msg)),
                device.getMetadata(),
                dataService.queryDeviceData(productId, msg.getDeviceId())
            ).doOnNext(tp2 -> {
                DeviceLatestData latestData = tp2.getT3();
                DeviceMetadata metadata = tp2.getT2();
                Map<String, Object> properties = new HashMap<>(tp2.getT1());
                for (PropertyMetadata property : metadata.getProperties()) {
                    properties.putIfAbsent(property.getId(), latestData.get(property.getId()));
                }
                DeviceMessageUtils.trySetProperties(msg, properties);
            })
            .onErrorResume((err) -> {
                log.error(err.getMessage(), err);
                return Mono.empty();
            })
            .thenReturn(msg);
    }
}
